package lhb.spark.sparkstreaming
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

class streamDemo {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(2))

    val groupID="testS"
    //kafka消费者参数
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "192.168.0.201:9092,192.168.0.202:9092,192.168.0.202:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> groupID,
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("testStream")

    val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      //任务均匀分布在excutor
      LocationStrategies.PreferConsistent,
      //订阅主题
      ConsumerStrategies.Subscribe(topics, kafkaParams))

    val inputValue: DStream[String] = dstream.map(tp => tp.value())
    //测试将dstream保存在文件中
    inputValue.saveAsTextFiles("./spark/data")

    ssc.start()
    //阻塞
    ssc.awaitTermination()
  }
}
